package com.sn.flink.streaming;

import lombok.AllArgsConstructor;
import lombok.NoArgsConstructor;
import lombok.ToString;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

public class SocketWindowWordCountJava {

    public static void main(String[] args) throws Exception {
        String hostname = "snzigod";
        int port;
        try {
            ParameterTool parameterTool = ParameterTool.fromArgs(args);
            port = parameterTool.getInt("port");
        } catch (Exception e) {
            System.err.println("未指定port，使用默认端口：9000");
            port = 9000;
        }
        //获取运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //获取socket数据
        DataStreamSource<String> text = env.socketTextStream(hostname, port, "\n");
        //编写算法
        text.flatMap(new FlatMapFunction<String, WordWithCount>() {
            public void flatMap(String value, Collector<WordWithCount> out) throws Exception {
                String[] splits = value.split("\\s");
                for (String word : splits) {
                    out.collect(new WordWithCount(word, 1L));
                }
            }
        }).keyBy("word").timeWindow(Time.seconds(2), Time.seconds(1)).sum("count").print().setParallelism(1);
        //flink是懒加载的，必须调用execute方法运算才会执行才会执行
        env.execute("streaming word count");
    }

    @ToString
    @NoArgsConstructor
    @AllArgsConstructor
    public static class WordWithCount {
        public String word;
        public long count;
    }

}
